package io.kiki.task.client.job.core.executor;

import io.kiki.task.client.job.core.MapHandler;
import io.kiki.task.client.job.core.cache.JobExecutorInfoCache;
import io.kiki.task.client.job.core.dto.JobExecutorInfo;
import io.kiki.task.client.job.core.dto.MapArgs;
import io.kiki.task.client.job.core.dto.MergeReduceArgs;
import io.kiki.task.client.job.core.dto.ReduceArgs;
import io.kiki.task.client.model.ExecuteResult;
import io.kiki.task.common.core.exception.JobMapReduceException;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.util.Objects;

/**
 * 基于注解的MapReduce执行器
 */
@Component
public class AnnotationMapReduceJobExecutor extends AbstractMapReduceExecutor {

    @Override
    protected ExecuteResult doReduceExecute(final ReduceArgs reduceArgs) {
        JobExecutorInfo jobExecutorInfo = JobExecutorInfoCache.get(reduceArgs.getExecutorInfo());
        if (Objects.isNull(jobExecutorInfo)) {
            throw new JobMapReduceException("[{}] not found", reduceArgs.getExecutorInfo());
        }

        if (Objects.isNull(jobExecutorInfo.getReduceExecutor())) {
            throw new JobMapReduceException(
                    "[{}] MapTask execution method not found. Please configure the @ReduceExecutor annotation",
                    reduceArgs.getExecutorInfo());
        }

        Class<?>[] paramTypes = jobExecutorInfo.getReduceExecutor().getParameterTypes();
        if (paramTypes.length > 0) {
            return (ExecuteResult) ReflectionUtils.invokeMethod(jobExecutorInfo.getReduceExecutor(),
                    jobExecutorInfo.getExecutor(), reduceArgs);
        }

        throw new JobMapReduceException("[{}] ReduceTask execution method not found", reduceArgs.getExecutorInfo());
    }

    @Override
    protected ExecuteResult doMergeReduceExecute(final MergeReduceArgs mergeReduceArgs) {
        JobExecutorInfo jobExecutorInfo = JobExecutorInfoCache.get(mergeReduceArgs.getExecutorInfo());

        if (Objects.isNull(jobExecutorInfo)) {
            throw new JobMapReduceException("[{}] not found", mergeReduceArgs.getExecutorInfo());
        }

        Method mergeReduceExecutor = jobExecutorInfo.getMergeReduceExecutor();
        if (Objects.isNull(mergeReduceExecutor)) {
            throw new JobMapReduceException(
                    "[{}] MapTask execution method not found. Please configure the @MergeReduceExecutor annotation",
                    mergeReduceArgs.getExecutorInfo());
        }

        Class<?>[] paramTypes = mergeReduceExecutor.getParameterTypes();
        if (paramTypes.length > 0) {
            return (ExecuteResult) ReflectionUtils.invokeMethod(mergeReduceExecutor,
                    jobExecutorInfo.getExecutor(), mergeReduceArgs);
        }

        throw new JobMapReduceException("[{}] MergeReduceTask execution method not found [{}]",
                mergeReduceArgs.getExecutorInfo());

    }

    @Override
    public ExecuteResult doJobMapExecute(final MapArgs mapArgs, final MapHandler mapHandler) {
        return invokeMapExecute(mapArgs, mapHandler);
    }
}
